package com.ds.infrastructure.hyperspace.container.internal.http2;

import com.ds.infrastructure.hyperspace.common.config.AbstractProtocolConfig;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundBuffer;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.http2.*;
import io.netty.handler.logging.LogLevel;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import static com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2FlowControllerState.DEFAULT_FLOW_CONTROLLER_STATE;
import static com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2FrameTypeEnum.*;
import static io.netty.buffer.ByteBufUtil.hexDump;
import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
import static io.netty.handler.codec.http2.Http2CodecUtil.connectionPrefaceBuf;
import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
import static io.netty.handler.codec.http2.Http2Exception.connectionError;
import static java.lang.Integer.MAX_VALUE;
import static java.lang.Math.min;

/**
 * @author  Weishaoying
 */
@Slf4j
public class HyperspaceHttp2Codec extends ByteToMessageDecoder {

    private static final ByteBuf HTTP_1_X_BUF = Unpooled.unreleasableBuffer(Unpooled.wrappedBuffer(
            new byte[]{'H', 'T', 'T', 'P', '/', '1', '.'})).asReadOnly();

    private static final int SETTING_MIN_SIZE = 5;

    private static final int SETTING_INITIAL_WINDOW_SIZE = 1048576;

    public static final int MAX_FRAME_SIZE_UPPER_BOUND = 0xffffff;

    private ByteBuf prefaceByteBuf;

    private final boolean isServer;

    private final Http2FrameReader frameReader;

    private final Http2FrameWriter frameWriter;

    private final Http2FrameListener readFrameListener;

    private final AbstractProtocolConfig protocolConfig;

    private boolean prefaceSent;

    private boolean isPrefaceDecoded;

    private List<ByteBuf> inList = new LinkedList<>();

    public HyperspaceHttp2Codec(boolean isServer, AbstractProtocolConfig protocolConfig) {
        this.isServer = isServer;
        this.readFrameListener = new HyperspaceHttp2FrameReadListener();
        if (isServer) {
            this.prefaceByteBuf = Http2CodecUtil.connectionPrefaceBuf();
        }
        if (log.isDebugEnabled()) {
            Http2FrameLogger frameLogger = new Http2FrameLogger(LogLevel.DEBUG,
                    HyperspaceHttp2FrameReadListener.class);
            this.frameReader = new Http2InboundFrameLogger(new DefaultHttp2FrameReader(), frameLogger);
            this.frameWriter = new Http2OutboundFrameLogger(new DefaultHttp2FrameWriter(), frameLogger);
        } else {
            this.frameReader = new DefaultHttp2FrameReader();
            this.frameWriter = new DefaultHttp2FrameWriter();
        }
        this.protocolConfig = protocolConfig;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        sendPreface(ctx, protocolConfig);
        ctx.fireChannelActive();
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        //inList.add(in);
        try {
            if (isPrefaceDecoded) {
                frameReader.readFrame(ctx, in, readFrameListener);
                return;
            }
            if (readClientPrefaceString(in) && verifyFirstFrameIsSettings(in)) {
                isPrefaceDecoded = true;
                frameReader.readFrame(ctx, in, readFrameListener);
            }
        } catch (Exception e) {
            log.error("Decode http2 message failed!", e);
            ctx.fireExceptionCaught(e);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        Iterator<ByteBuf> iterator = inList.iterator();
        while (iterator.hasNext()) {
            ByteBuf data = iterator.next();
            if (data.refCnt() == 0) {
                iterator.remove();
            }
        }
        super.channelReadComplete(ctx);
    }

    @Override
    public void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
        for (ByteBuf byteBuf : inList) {
            if (byteBuf.refCnt() > 0) {
                log.error(String.format("HTTP2:byteBuf.refCnt()>0:%d;inList.size()%d;channelId:%s;%s",
                        byteBuf.refCnt(), inList.size(), ctx.channel().id(), byteBuf.isReadable()));
                byteBuf.release(byteBuf.refCnt());
                byteBuf.clear();
            }
        }
        ChannelOutboundBuffer channelOutboundBuffer = ctx.channel().unsafe().outboundBuffer();
        if (channelOutboundBuffer != null) {
            for (; ; ) {
                if (!channelOutboundBuffer.remove()) {
                    break;
                }
            }
        }
    }

    public void writeAndFlushSettingsAck(ChannelHandlerContext ctx) {
        frameWriter.writeSettingsAck(ctx, ctx.newPromise());
        ctx.flush();
    }

    public void writeAndFlushWindowUpdateFrame(ChannelHandlerContext ctx, int windowUpdateIncrement) {
        HyperspaceHttp2FrameWrapper http2WindowUpdateFrameWrapper =
                new HyperspaceHttp2FrameWrapper(WINDOW_UPDATE, new DefaultHttp2WindowUpdateFrame(windowUpdateIncrement));
        writeAndFlushData(ctx, http2WindowUpdateFrameWrapper);
    }

    public void applySettings(Http2Settings settings) {
        applyFrameReaderSettingsInternal(settings, frameReader.configuration());
        applyFrameWriterSettingsInternal(settings, frameWriter.configuration());
    }

    public ChannelPromise writeData(ChannelHandlerContext channelHandlerContext, Object msg) {
        if (!channelHandlerContext.channel().isWritable()) {
            channelHandlerContext.flush();
        }
        ChannelPromise promise = channelHandlerContext.newPromise().addListener(FIRE_EXCEPTION_ON_FAILURE);
        try {
            if (!(msg instanceof HyperspaceHttp2FrameWrapper)) {
                channelHandlerContext.write(msg, promise);
                return promise;
            }
            HyperspaceHttp2FrameWrapper frameWrapper = (HyperspaceHttp2FrameWrapper) msg;
            HyperspaceHttp2FrameTypeEnum frameType = frameWrapper.getFrameType();
            Object frameData = frameWrapper.getFrameData();
            if (DATA == frameType) {
                Http2DataFrame dataFrame = (Http2DataFrame) frameData;
                frameWriter.writeData(channelHandlerContext, frameWrapper.getStreamId(), dataFrame.content(),
                        dataFrame.padding(), dataFrame.isEndStream(), promise);
            } else if (HEADERS == frameType) {
                Http2HeadersFrame headersFrame = (Http2HeadersFrame) frameData;
                HyperspaceHttp2FlowControllerState flowControllerState = frameWrapper.getFlowControllerState();
                if (flowControllerState != null) {
                    frameWriter.writeHeaders(channelHandlerContext, frameWrapper.getStreamId(),
                            headersFrame.headers(), flowControllerState.getStreamDependency(),
                            flowControllerState.getWeight(), flowControllerState.isExclusive(),
                            headersFrame.padding(), headersFrame.isEndStream(), promise);
                } else {
                    frameWriter.writeHeaders(channelHandlerContext, frameWrapper.getStreamId(),
                            headersFrame.headers(), headersFrame.padding(),
                            headersFrame.isEndStream(), promise);
                }

            } else if (WINDOW_UPDATE == frameType) {
                Http2WindowUpdateFrame windowUpdateFrame = (Http2WindowUpdateFrame) frameData;
                frameWriter.writeWindowUpdate(channelHandlerContext, frameWrapper.getStreamId(),
                        windowUpdateFrame.windowSizeIncrement(), promise);
            } else if (RST_STREAM == frameType) {
                Http2ResetFrame rstFrame = (Http2ResetFrame) frameData;
                frameWriter.writeRstStream(channelHandlerContext,
                        frameWrapper.getStreamId(), rstFrame.errorCode(), promise);
            } else if (PING == frameType) {
                Http2PingFrame frame = (Http2PingFrame) frameData;
                frameWriter.writePing(channelHandlerContext, frame.ack(), frame.content(), promise);
            } else if (PING_ACK == frameType) {
                Http2PingFrame frame = (Http2PingFrame) frameData;
                frameWriter.writePing(channelHandlerContext, true, frame.content(), promise);
            } else if (SETTINGS == frameType) {
                frameWriter.writeSettings(channelHandlerContext, ((Http2SettingsFrame) frameData).settings(),
                        promise);
            } else if (SETTINGS_ACK == frameType) {
                frameWriter.writeSettingsAck(channelHandlerContext, promise);
            } else if (PRIORITY == frameType) {
                HyperspaceHttp2FlowControllerState flowControllerState = frameWrapper.getFlowControllerState();
                if (flowControllerState == null) {
                    flowControllerState = DEFAULT_FLOW_CONTROLLER_STATE;
                }
                frameWriter.writePriority(channelHandlerContext, frameWrapper.getStreamId(),
                        flowControllerState.getStreamDependency(), flowControllerState.getWeight(),
                        flowControllerState.isExclusive(), promise);
            } else if (GO_AWAY == frameType) {
                HyperspaceHttp2GoAwayFrame hyperspaceHttp2GoAwayFrame = (HyperspaceHttp2GoAwayFrame) frameData;
                Http2GoAwayFrame originGoAwayFrame = hyperspaceHttp2GoAwayFrame.getOriginGoAwayFrame();
                frameWriter.writeGoAway(channelHandlerContext, hyperspaceHttp2GoAwayFrame.getLastStreamId(),
                        originGoAwayFrame.errorCode(), originGoAwayFrame.content(), promise);
            } else if (UNKNOWN == frameType) {
                Http2UnknownFrame unknownFrame = (Http2UnknownFrame) frameData;
                frameWriter.writeFrame(channelHandlerContext, unknownFrame.frameType(), frameWrapper.getStreamId(),
                        unknownFrame.flags(), unknownFrame.content(), promise);
            } else {
                log.warn("Unsupport http2 data type {}, data {}. ", frameType, msg);
                ReferenceCountUtil.release(frameData);
            }
        } catch (Exception e) {
            log.error("Write http2 message failed!", e);
        }
        return promise;
    }

    public ChannelPromise writeAndFlushData(ChannelHandlerContext channelHandlerContext, Object msg) {
        ChannelPromise channelPromise = writeData(channelHandlerContext, msg);
        channelHandlerContext.flush();
        return channelPromise;
    }

    public void sendPreface(ChannelHandlerContext ctx, AbstractProtocolConfig protocolConfig) {
        if (prefaceSent) {
            return;
        }
        prefaceSent = true;
        if (isServer) {
            Http2Settings settings = new Http2Settings();
            settings.pushEnabled(false);
            settings.initialWindowSize(SETTING_INITIAL_WINDOW_SIZE);
            settings.maxConcurrentStreams(MAX_VALUE);
            settings.maxHeaderListSize(protocolConfig.getMaxHeaderSize());
            frameWriter.writeSettings(ctx, settings, ctx.newPromise()).addListener(
                    ChannelFutureListener.CLOSE_ON_FAILURE);
        } else {
            ctx.write(connectionPrefaceBuf())
                    .addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
        }
        ctx.flush();
    }

    private boolean readClientPrefaceString(ByteBuf in) throws Http2Exception {
        if (prefaceByteBuf == null) {
            return true;
        }

        int prefaceRemaining = prefaceByteBuf.readableBytes();
        int bytesRead = min(in.readableBytes(), prefaceRemaining);
        if (bytesRead == 0 || !ByteBufUtil.equals(in, in.readerIndex(),
                prefaceByteBuf, prefaceByteBuf.readerIndex(), bytesRead)) {
            int maxSearch = 1024;
            int http1Index = ByteBufUtil.indexOf(HTTP_1_X_BUF, in.slice(in.readerIndex(), min(in.readableBytes(), maxSearch)));
            if (http1Index != -1) {
                String chunk = in.toString(in.readerIndex(), http1Index - in.readerIndex(), CharsetUtil.US_ASCII);
                throw connectionError(PROTOCOL_ERROR, "Unexpected HTTP/1.x request: %s", chunk);
            }
            String receivedBytes = hexDump(in, in.readerIndex(),
                    min(in.readableBytes(), prefaceByteBuf.readableBytes()));
            throw connectionError(PROTOCOL_ERROR, "HTTP/2 client preface string missing or corrupt. " +
                    "Hex dump for received bytes: %s", receivedBytes);
        }
        in.skipBytes(bytesRead);
        prefaceByteBuf.skipBytes(bytesRead);
        if (!prefaceByteBuf.isReadable()) {
            // Entire preface has been read.
            prefaceByteBuf.release();
            prefaceByteBuf = null;
            return true;
        }
        return false;
    }

    private boolean verifyFirstFrameIsSettings(ByteBuf in) throws Http2Exception {
        if (in.readableBytes() < SETTING_MIN_SIZE) {
            return false;
        }

        short frameType = in.getUnsignedByte(in.readerIndex() + 3);
        short flags = in.getUnsignedByte(in.readerIndex() + 4);
        if (frameType != Http2FrameTypes.SETTINGS || (flags & Http2Flags.ACK) != 0) {
            throw connectionError(PROTOCOL_ERROR, "First received frame was not SETTINGS. " +
                            "Hex dump for first 5 bytes: %s",
                    hexDump(in, in.readerIndex(), 5));
        }
        return true;
    }

    private void applyFrameReaderSettingsInternal(Http2Settings settings, Http2FrameReader.Configuration config) {
        try {
            final Http2HeadersDecoder.Configuration headerConfig = config.headersConfiguration();
            final Http2FrameSizePolicy frameSizePolicy = config.frameSizePolicy();

            Long headerTableSize = settings.headerTableSize();
            if (headerTableSize != null) {
                headerConfig.maxHeaderTableSize(headerTableSize);
            }

            Long maxHeaderListSize = settings.maxHeaderListSize();
            if (maxHeaderListSize != null) {
                headerConfig.maxHeaderListSize(maxHeaderListSize,
                        Http2CodecUtil.calculateMaxHeaderListSizeGoAway(maxHeaderListSize));
            }

//            issue : io.netty.handler.codec.http2.Http2Exception: Frame length: 43078 exceeds maximum: 16384
//            Integer maxFrameSize = settings.maxFrameSize();
//            if (maxFrameSize != null) {
//                frameSizePolicy.maxFrameSize(maxFrameSize);
//            }
            frameSizePolicy.maxFrameSize(MAX_FRAME_SIZE_UPPER_BOUND);
        } catch (Http2Exception e) {
            log.error("Apply setting frame reader failed.", e);
        }
    }

    private void applyFrameWriterSettingsInternal(Http2Settings settings, Http2FrameWriter.Configuration config) {
        try {
            Http2HeadersEncoder.Configuration outboundHeaderConfig = config.headersConfiguration();
            Http2FrameSizePolicy outboundFrameSizePolicy = config.frameSizePolicy();
            Long headerTableSize = settings.headerTableSize();
            if (headerTableSize != null) {
                outboundHeaderConfig.maxHeaderTableSize((int) min(headerTableSize, MAX_VALUE));
            }

            Long maxHeaderListSize = settings.maxHeaderListSize();
            if (maxHeaderListSize != null) {
                outboundHeaderConfig.maxHeaderListSize(maxHeaderListSize);
            }
//            issue : io.netty.handler.codec.http2.Http2Exception: Frame length: 43078 exceeds maximum: 16384
//            Integer maxFrameSize = settings.maxFrameSize();
//            if (maxFrameSize != null) {
//                outboundFrameSizePolicy.maxFrameSize(maxFrameSize);
//            }
            outboundFrameSizePolicy.maxFrameSize(MAX_FRAME_SIZE_UPPER_BOUND);
        } catch (Http2Exception e) {
            log.error("Apply setting to frame writer failed.", e);
        }
    }

}
